#include <iostream>
// #include <asio.hpp>
#include "test.pb.h"
#include "test.grpc.pb.h"
// #include <grpc++/grpc++.h>
// #include "function_traits.hpp"
#include "proto_service.hpp"
#include <asio/thread_pool.hpp>
#include <optional>
#include <asio/redirect_error.hpp>
#include <asio/steady_timer.hpp>

asio::awaitable<void> stream_return(meta_data* storage,
                                    const int_message& req,
                                    std::function<asio::awaitable<void>(std::optional<int_message>)> stream_response)
{
    // std::cout << "stream process started "<< std::endl;
    auto timer = asio::steady_timer(co_await asio::this_coro::executor);
    asio::error_code c;
    for(int i=0;i<3;++i)
    {
        timer.expires_after(std::chrono::seconds{1});
        co_await timer.async_wait(asio::redirect_error(asio::use_awaitable,c));
        int_message r;
        r.set_a(i);
        std::cout << "returns  "<<i<< std::endl;
        co_await stream_response(r);
    }
    std::cout << "finish  stream call"<< std::endl;
    // co_await stream_response({});
    // std::cout << "finished"<< std::endl;
    co_return;
}

asio::awaitable<int_message> stream_param(meta_data* storage,std::function<asio::awaitable<std::optional<int_message>>()> param)
{
    int_message r;
    int base = 0;
    while(auto r = co_await param())
    {
        base += r->a();
        std::cout<<"recive "<<r->a()<<std::endl;
    }
    r.set_a(base);
    co_return r;
}

asio::awaitable<void> stream_all(meta_data* storage,std::function<asio::awaitable<std::optional<int_message>>()> param,
                                std::function<asio::awaitable<void>(std::optional<int_message>)> stream_response)
{
    for(auto ite : storage->client_data){
        std::cout<<"client meatadata ["<<ite.first<<"] : "<<ite.second<<std::endl;
    }
    auto timer = asio::steady_timer(co_await asio::this_coro::executor);
    asio::error_code c;
    while(auto r = co_await param())
    {
        timer.expires_after(std::chrono::seconds{1});
        co_await timer.async_wait(asio::redirect_error(asio::use_awaitable,c));
        std::cout<<"response to client "<<r->a()<<std::endl;
        co_await stream_response(*r);
    }
    // std::cout<<"await over "<<std::endl;
    // throw grpc::Status::CANCELLED; 
    co_return;
}

asio::awaitable<int_message> fuck_add(meta_data* storage,const add_message &req)
{
    int_message r;
    r.set_a(req.a()+req.b());
    co_return std::move(r);
}
asio::awaitable<int_message> fuck(meta_data* storage,const int_message &req)
{
    std::cout << "process started "<< std::endl;
    int_message r;
    asio::thread_pool p;
    auto i = co_await asio::co_spawn(
        p, []() -> asio::awaitable<int> {
            std::cout << "sleep started "<< std::endl;
            std::this_thread::sleep_for(std::chrono::seconds(1));
            std::cout << "sleep ended "<< std::endl;
            co_return 100;
        },
        asio::use_awaitable);
    r.set_a(req.a() + i);
    std::cout << "process ended "<< std::endl;
    co_return std::move(r);
}
asio::awaitable<int_message> fuck_sub(meta_data* storage,const add_message &req)
{
    int_message r;
    r.set_a(req.a()-req.b());
    co_return std::move(r);
}
asio::awaitable<void> test_middle_ware(meta_data*)
{
    std::cout<<"middleware triggered"<<std::endl;
    throw grpc::Status(grpc::StatusCode::PERMISSION_DENIED,"un authed action");
    co_return;
}
typedef GRpcServerContext<test::AsyncService,test2::AsyncService,test3::AsyncService> CType;
int main()
{
    asio::io_context io;
    auto& context = asio::use_service<CType>(io);
    context.listen(50051);
    context.use_middle_ware(test_middle_ware);
    context.bind(&test::AsyncService::Requestfuck, fuck);
    context.bind(&test::AsyncService::Requestfuck_add, fuck_add);
    context.bind(&test2::AsyncService::Requestfuck_sub, fuck_sub);
    context.bind(&test3::AsyncService::Requeststream_return, stream_return);
    context.bind(&test3::AsyncService::Requeststream_param, stream_param);
    context.bind(&test3::AsyncService::Requeststream_test, stream_all);
    auto guard = asio::make_work_guard(io);
    io.run();
    return 0;
}